package s_processPkg

import (
	"Infinichat_mixed/back/mapper"
	"Infinichat_mixed/entity"
	"Infinichat_mixed/server/s_service"
	"Infinichat_mixed/server/s_utils"
	"Infinichat_mixed/transfer"
	"encoding/json"
	"fmt"
	"github.com/redis/go-redis/v9"
	"net"
	"slices"
	"strconv"
	"time"
)

type UserProcess struct {
	Conn   net.Conn
	UserId uint // 表明这个Conn是哪个用户的
}

func (this *UserProcess) Login(mes *entity.Message) (err error) {
	var loginMes entity.LoginMes
	err = json.Unmarshal([]byte(mes.Data), &loginMes)
	if err != nil {
		fmt.Println("Unmarshal err:", err)
		return
	}

	var loginResMes entity.LoginResMes
	user, sessionToken, err := s_service.Login(loginMes.UserAccount, loginMes.UserPwd)

	if err != nil {
		if err == s_utils.ERROR_USER_NOTEXISTS {
			loginResMes.Code = 500
			loginResMes.Error = err.Error()
		} else if err == s_utils.ERROR_USER_PWD {
			loginResMes.Code = 300
			loginResMes.Error = err.Error()
		} else if err == s_utils.ERROR_USER_LOGINED {
			loginResMes.Code = 300
			loginResMes.Error = err.Error()
		} else {
			loginResMes.Code = 505
			loginResMes.Error = "server error."
		}
		fmt.Println("Login err:", err)
	} else {
		this.UserId = user.ID
		OnlineUserMgr.UserTokens[user.ID] = sessionToken
		OnlineUserMgr.AddOnlineUser(this)
		loginResMes.Code = 200
		loginResMes.User = user
		status := entity.UserOnline
		this.NotifyOthers(user.ID, status)
		users := OnlineUserMgr.OnlineUsers
		userFriendIds := s_service.GetFriend(user.ID)
		for id, _ := range users {
			// 只有是好友才显示
			if id != user.ID && slices.Contains(userFriendIds, id) {
				loginResMes.UserIds = append(loginResMes.UserIds, id)
			}
		}
		// 加载好友数据进去redis，以实现共同好友
		LoadFriends(user.ID)

		// 登录后要查看是否有未读消息 好友申请 群聊邀请
		p2p_msg, p2m_msg, addIds, gis, _ := s_service.GetReq(user.ID)
		loginResMes.P2p_msg = p2p_msg
		loginResMes.P2m_msg = p2m_msg
		loginResMes.AddIds = addIds
		loginResMes.Gis = gis

		// 设置最后活跃时间
		SetLastActive(user.ID)
	}
	data, _ := json.Marshal(loginResMes)
	var resMes entity.Message
	resMes.Type = entity.LoginResMesType
	resMes.Data = string(data)
	data, _ = json.Marshal(resMes)
	tf := &transfer.Transfer{this.Conn}
	_ = tf.WriterPkg(data)
	return err
}

func (this *UserProcess) Logout(id uint) {
	// 从用户在线列表删除
	delete(OnlineUserMgr.OnlineUsers, id)
	// 通知其他用户 id outline
	//status := message.UserOffline
	//this.NotifyOthers(id, status)
	sessionToken := OnlineUserMgr.UserTokens[id]
	s_service.ReleaseLoginLock(id, sessionToken)
}

func (this *UserProcess) NotifyOthers(userId uint, status uint8) {
	// 非好友不通知
	notify_data := this.constructMsg(userId, status)
	userFriendIds := s_service.GetFriend(userId)

	for id, up := range OnlineUserMgr.OnlineUsers {
		if id == userId || !slices.Contains(userFriendIds, id) {
			continue
		} else {
			up.NotifyMeStauts(notify_data)
		}
	}
}

func (this *UserProcess) constructMsg(userId uint, status uint8) (data []byte) {
	var mes entity.Message
	mes.Type = entity.NotifyUserStatusMesType

	var notifyUserStatusMes entity.NotifyUserStatusMes
	notifyUserStatusMes.UserId = userId
	//notifyUserStatusMes.Status = message.UserOnline
	notifyUserStatusMes.Status = status

	data, err := json.Marshal(notifyUserStatusMes)
	if err != nil {
		fmt.Println("Marshal err:", err)
		return
	}

	mes.Data = string(data)
	data, err = json.Marshal(mes)
	if err != nil {
		fmt.Println("Marshal err:", err)
		return
	}

	return data
}

func (this *UserProcess) NotifyMeStauts(data []byte) {

	tf := &transfer.Transfer{this.Conn}
	err := tf.WriterPkg(data)
	if err != nil {
		fmt.Println("NotifyMeStauts err:", err)
		return
	}
}

//// 申请添加好友：直接添加到mysql中
//// 如果好友离线，则需要在其上线后推送 ==> 延迟到登录那块去做
//func (this *UserProcess) AskAdd(mes *message.Message) (err error) {
//	var afa message.AskForAdd
//	err = json.Unmarshal([]byte(mes.Data), &afa)
//	if err != nil {
//		return err
//	}
//	sid := afa.User.ID
//	tid := afa.TagetId
//	err = s_service.AskAdd(sid, tid)
//	if err != nil {
//		return err
//	}
//	// 如果好友在线，则需要实时推送
//	//tf := &transfer.Transfer{this.Conn, [8192]byte{}}
//	//data, err := json.Marshal(mes)
//	//err = tf.WriterPkg(data)
//	return err
//}

func LoadFriends(uid uint) {
	// 先看看redis中有没有
	if !IsFriendsInRedis(uid) {
		// 不存在redis中
		// 从mysql中加载，设置过期时间为 1h
		retIds, _ := LoadFriedsFromMysql(uid)
		_ = AddFriends2Redis(uid, retIds)
	} else {
		// redis中有数据 更新过期时间
		//conn := s_utils.RedisPool.Get()
		//defer conn.Close()
		//key := fmt.Sprintf("friends:%d", uid)
		//conn.Send("EXPIRE", key, 3600)
		_, _ = s_utils.RedisClient.Do(s_utils.Ctx, "EXPIRE", fmt.Sprintf("friends:%d", uid), 3600).Result()
	}
}

func LoadFriedsFromMysql(uid uint) (retIds []uint, err error) {
	var friends []entity.Friend
	s_utils.DB.Select("sid, tid").Where("sid = ? or tid = ?", uid, uid).Find(&friends)
	for _, friend := range friends {
		if friend.Sid == uid {
			retIds = append(retIds, friend.Tid)
		} else if friend.Tid == uid {
			retIds = append(retIds, friend.Sid)
		}
	}
	return retIds, nil
}

func IsFriendsInRedis(uid uint) bool {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	key := fmt.Sprintf("friends:%d", uid)
	//values, err := redis.Values(conn.Do("SMEMBERS", key))
	//values, err := redis.Values(s_utils.RedisClient.Do(s_utils.Ctx, "SMEMBERS", key).Result())
	values, err := s_utils.RedisClient.SMembers(s_utils.Ctx, key).Result()
	if err != nil || len(values) == 0 {
		return false
	}
	return true
}

func AddFriends2Redis(uid uint, friends []uint) error {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	redisClient := s_utils.RedisClient
	ctx := s_utils.Ctx
	key := fmt.Sprintf("friends:%d", uid)
	// DEL friends:1
	//conn.Send("DEL", key)
	_, _ = redisClient.Del(ctx, key).Result()
	//args := redis.Args{}.Add(key).AddFlat(friends)
	//// SADD friends:1 2 3 10 11 12 13 50
	//conn.Send("SADD", args...)
	//conn.Send("EXPIRE", key, 3600)
	friendsInterface := make([]interface{}, len(friends))
	for i, v := range friends {
		friendsInterface[i] = v
	}
	// 使用 RedisClient 进行 SADD 操作
	_, _ = redisClient.SAdd(ctx, key, friendsInterface...).Result()

	_, _ = redisClient.Expire(ctx, key, 3600*time.Second).Result()
	return nil
}

func GetMutualFriends(userA, userB uint) ([]uint, error) {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	keyA := fmt.Sprintf("friends:%d", userA)
	keyB := fmt.Sprintf("friends:%d", userB)
	//res, _ := redis.Strings(conn.Do("SINTER", keyA, keyB))
	// 使用 redisClient 执行 SINTER 操作
	// 使用 s_util.RedisClient 和 s_util.Ctx 执行 SINTER 操作
	res, _ := s_utils.RedisClient.SInter(s_utils.Ctx, keyA, keyB).Result()
	uids := make([]uint, len(res))
	for k, v := range res {
		uid, _ := strconv.ParseUint(v, 10, 64)
		uids[k] = uint(uid)
	}
	return uids, nil
}

// ZRANGE messages:1min 0 -1 WITHSCORES 可以获取ZSET中的数据
func PushDelayedMessage(TimeSend entity.TimeSend) error {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	delay := TimeSend.Time * time.Second
	sendAt := time.Now().Add(delay).Unix()
	var zsetKey string
	if delay <= time.Minute {
		zsetKey = "messages:1min"
	} else if delay <= 10*time.Minute {
		zsetKey = "messages:10min"
	} else if delay <= time.Hour {
		zsetKey = "messages:1h"
	} else {
		zsetKey = "messages:other"
	}
	message, _ := json.Marshal(TimeSend)
	//_, err := conn.Do("ZADD", zsetKey, sendAt, string(message))
	_, err := s_utils.RedisClient.ZAdd(s_utils.Ctx, zsetKey, redis.Z{
		Score:  float64(sendAt),
		Member: message,
	}).Result()
	if err != nil {
		fmt.Println("=====================")
	}
	return err
}

func ProcessDelayedMessages() {
	//ticker := time.NewTicker(time.Second * 1)
	//defer ticker.Stop()
	//for {
	//	select {
	//	case <-ticker.C:
	//		now := time.Now().Unix()
	//		processMessages(conn, "messages:1min", now)
	//		processMessages(conn, "messages:10min", now)
	//		processMessages(conn, "messages:1h", now)
	//	}
	//}
	go process1MinQueue()
	go process10MinQueue()
	go process1hQueue()
	select {}
}

func process1MinQueue() {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	ticker := time.NewTicker(time.Second * 1)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			now := time.Now().Unix()
			processMessages("messages:1min", now)
		}
	}
}

func process10MinQueue() {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			now := time.Now().Unix()
			//msgs, err := redis.Strings(conn.Do("ZRANGEBYSCORE", "messages:10min", "-inf", now))
			msgs, err := s_utils.RedisClient.ZRangeByScore(s_utils.Ctx, "messages:10min", &redis.ZRangeBy{
				Min: "-inf",
				Max: strconv.FormatFloat(float64(now), 'f', -1, 64),
			}).Result()
			if err != nil {
				fmt.Println("ZRANGEBYSCORE error:", err)
				return
			}
			for _, msg := range msgs {
				var ts entity.TimeSend
				_ = json.Unmarshal([]byte(msg), &ts)
				sendAt := time.Now().Add(time.Duration(ts.Time) * time.Second).Unix()

				if sendAt-now <= 60 {
					//_, _ = conn.Do("ZREM", "messages:10min", msg)
					//_, _ = conn.Do("ZADD", "messages:1min", sendAt, msg)
					_, _ = s_utils.RedisClient.ZRem(s_utils.Ctx, "messages:10min", msg).Result()
					_, _ = s_utils.RedisClient.ZAdd(s_utils.Ctx, "messages:1min", redis.Z{
						Score:  float64(sendAt),
						Member: msg,
					}).Result()
				}
			}
		}
	}
}

func process1hQueue() {
	//conn := s_utils.RedisPool.Get()
	//defer conn.Close()
	ticker := time.NewTicker(10 * time.Minute)
	defer ticker.Stop()
	for range ticker.C {
		now := time.Now().Unix()

		targetTime := now + 600 // 10分钟以内
		//msgs, err := redis.Strings(conn.Do("ZRANGEBYSCORE", "messages:1h", "-inf", targetTime))
		msgs, err := s_utils.RedisClient.ZRangeByScore(s_utils.Ctx, "messages:1h", &redis.ZRangeBy{
			Min: "-inf",
			Max: strconv.FormatFloat(float64(targetTime), 'f', -1, 64),
		}).Result()
		if err != nil {
			fmt.Println("ZRANGEBYSCORE error:", err)
			return
		}

		for _, msg := range msgs {
			var ts entity.TimeSend
			_ = json.Unmarshal([]byte(msg), &ts)
			sendAt := time.Now().Add(time.Duration(ts.Time) * time.Second).Unix()

			if sendAt-now <= 600 {
				_, _ = s_utils.RedisClient.ZRem(s_utils.Ctx, "messages:1h", msg).Result()
				_, _ = s_utils.RedisClient.ZAdd(s_utils.Ctx, "messages:10min", redis.Z{
					Score:  float64(sendAt),
					Member: msg,
				}).Result()

			}
		}
	}
}

func processMessages(zsetKey string, currentTime int64) {
	//msgs, err := redis.Strings(conn.Do("ZRANGEBYSCORE", zsetKey, "-inf", currentTime))
	msgs, _ := s_utils.RedisClient.ZRangeByScore(s_utils.Ctx, zsetKey, &redis.ZRangeBy{
		Min: "-inf",
		Max: fmt.Sprintf("%d", currentTime), // currentTime 是一个整数，需转换为字符串
	}).Result()
	for _, msg := range msgs {
		TimeSend := entity.TimeSend{}
		json.Unmarshal([]byte(msg), &TimeSend)
		sendMessage(TimeSend)
		//conn.Do("ZREM", zsetKey, msg)
		_, _ = s_utils.RedisClient.ZRem(s_utils.Ctx, zsetKey, msg).Result()
	}
}

func sendMessage(TimeSend entity.TimeSend) {
	smsProcess := &SmsProcess{}
	var mes = new(entity.Message)
	mes.Type = entity.SmsMes2OneType
	var smsMes2one entity.P2PMsg
	smsMes2one.ReceiveId = TimeSend.RecvId
	smsMes2one.Content = TimeSend.Msg
	smsMes2one.SendId = TimeSend.SendId
	data, _ := json.Marshal(smsMes2one)
	mes.Data = string(data)
	smsProcess.SendMes2One(mes)
}

func StartP2PMsgSync() {
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			go syncP2PMessages()
		}
	}
}

func syncP2PMessages() {
	fmt.Println("单点消息守护线程到来，准备刷数据进MySQL...")
	//redisConn := s_utils.RedisPool.Get()
	//defer redisConn.Close()

	timestamp := time.Now().Unix()
	//keys, _ := redis.Strings(redisConn.Do("SMEMBERS", "active:zsetKeys"))
	keys, _ := s_utils.RedisClient.SMembers(s_utils.Ctx, "active:zsetKeys").Result()
	for _, zsetKey := range keys {
		//msgData, _ := redis.Values(redisConn.Do("ZRANGEBYSCORE", zsetKey, "-inf", timestamp))
		msgData, _ := s_utils.RedisClient.ZRangeByScore(s_utils.Ctx, zsetKey, &redis.ZRangeBy{
			Min: "-inf",
			Max: fmt.Sprintf("%d", timestamp),
		}).Result()
		sid, tid, _ := entity.ExtractSidTidFromZsetKey(zsetKey)
		syncSetKey := entity.GetSyncSetKey(sid, tid)
		var msgsToInsert []entity.P2PMsg
		for _, md := range msgData {
			var msg entity.P2PMsg
			_ = json.Unmarshal([]byte(md), &msg)
			//exist, _ := redis.Bool(redisConn.Do("SISMEMBER", syncSetKey, msg.ID))
			exist, _ := s_utils.RedisClient.SIsMember(s_utils.Ctx, syncSetKey, msg.ID).Result()
			if !exist {
				msgsToInsert = append(msgsToInsert, msg)
			}
		}
		if len(msgsToInsert) > 0 {
			retMsgInsert, err := mapper.SaveP2PMessages(msgsToInsert)
			if err == nil {
				for _, msg := range retMsgInsert {
					//_, _ = redisConn.Do("SADD", syncSetKey, msg.ID)
					_, _ = s_utils.RedisClient.SAdd(s_utils.Ctx, syncSetKey, msg.ID).Result()
				}
			}
		}
		syncField := entity.GetSyncField(sid, tid)

		//// 从活跃会话集合中移除
		//// 这里有个问题，当守护线程在写入后，这个zset被从活跃线程中移除了，但又同时有数据写入，这样就会丢失数据
		//_, _ = redisConn.Do("SREM", "active:zsetKeys", zsetKey)
		go func(zsetKey, syncField string, timestamp int64) {
			time.Sleep(2 * time.Second)
			//_, _ = redisConn.Do("ZREMRANGEBYSCORE", zsetKey, -1, timestamp)
			//_, _ = redisConn.Do("SET", syncField, "false")
			_, _ = s_utils.RedisClient.ZRemRangeByScore(s_utils.Ctx, zsetKey, "-1", fmt.Sprintf("%d", timestamp)).Result()
			_, _ = s_utils.RedisClient.Set(s_utils.Ctx, syncField, "false", 0).Result()
		}(zsetKey, syncField, timestamp)
	}
}

func StartHeartbeatChecker() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	RedisZSetKey := "tcp:active_connections"
	for range ticker.C {
		now := time.Now().Unix()
		threshold := now - 30 // 30秒未活跃
		// 查找不活跃的连接
		connIds, _ := s_utils.RedisClient.ZRangeByScore(s_utils.Ctx, RedisZSetKey, &redis.ZRangeBy{
			Min: "0",
			Max: strconv.FormatFloat(float64(threshold), 'f', -1, 64),
		}).Result()
		for _, connID := range connIds {
			// 发送心跳包 PING
			uid, err := strconv.Atoi(connID)
			if err == nil {
				if conn := GetConnById(uint(uid)); conn != nil {
					go func(c net.Conn, id uint) {
						tf := &transfer.Transfer{Conn: conn}
						mes := &entity.Message{}
						mes.Type = entity.HeartBeatType
						data, _ := json.Marshal(mes)
						_ = tf.WriterPkg(data)
						select {
						case <-HeartBeatMp[id]:
							// 收到心跳回复
							// 重新设置最后活跃时间
							SetLastActive(id)
						case <-time.After(10 * time.Second):
							// 超时未响应，关闭连接
							delete(OnlineUserMgr.OnlineUsers, id)
							sessionToken := OnlineUserMgr.UserTokens[id]
							// 移除连接
							s_utils.RedisClient.ZRem(s_utils.Ctx, RedisZSetKey, connID)
							s_service.ReleaseLoginLock(id, sessionToken)
							_ = c.Close()
						}
					}(conn, uint(uid))
				}
			} else {
				fmt.Println("strconv.Atoi(connID)失败...")
			}
		}
	}
}

func SetLastActive(uid uint) {
	connID := strconv.Itoa(int(uid))
	RedisZSetKey := "tcp:active_connections"
	s_utils.RedisClient.ZAdd(s_utils.Ctx, RedisZSetKey, redis.Z{
		Score:  float64(time.Now().Unix()),
		Member: connID,
	})
}
